#pragma once
#include "../third/protobuf/codec.h"
#include "../third/protobuf/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "../mqcommon/ThreadPool.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/proto.pb.h"
#include "../mqcommon/Log.hpp"
#include "Connection.hpp"
#include "Consumer.hpp"
namespace mq
{
    class Server
    {
#define HOST_NAME "host"
#define DBFILE "/meta.db"
    public:
        Server(int port = 8088, const string &basedir = "./data")
            : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort),
              _dispatcher(bind(&Server::onUnknownMessage, this, placeholders::_1, placeholders::_2, placeholders::_3)),
              _codec(make_shared<ProtobufCodec>(bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, placeholders::_1, placeholders::_2, placeholders::_3))),
              _host(make_shared<VirtualHost>(HOST_NAME, basedir, basedir + DBFILE)),
              _consumer_manager(make_shared<ConsumerManager>()),
              _connection_manager(make_shared<ConnectionManager>()),
              _pool(make_shared<ThreadPool>())
        {
            MsgQueueManager::QueueMap qm = _host->allqueue();
            for (auto &it : qm)
                _consumer_manager->initQueueConsumer(it.first);
            _dispatcher.registerMessageCallback<mq::openChannelRequest>(bind(&Server::onOpenChannel, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::closeChannelRequest>(bind(&Server::onCloseChannel, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::declareExchangeRequest>(bind(&Server::onDeclareExchange, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::deleteExchangeRequest>(bind(&Server::onDeleteExchange, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::declareQueueRequest>(bind(&Server::onDeclareQueue, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::deleteQueueRequest>(bind(&Server::onDeleteQueue, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::BindRequest>(bind(&Server::onQueueBind, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::unBindRequest>(bind(&Server::onQueueUnBind, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicPublishRequest>(bind(&Server::onBasicPublish, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicAckRequest>(bind(&Server::onBasicAck, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicCancelRequest>(bind(&Server::onBasicCancel, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicConsumeRequest>(bind(&Server::onBasicConsume, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _server.setConnectionCallback(bind(&Server::onConnection, this, placeholders::_1));
            _server.setMessageCallback(bind(&ProtobufCodec::onMessage, _codec.get(), placeholders::_1, placeholders::_2, placeholders::_3));
        }
        void start()
        {
            _server.start();
            _baseloop.loop();
        }
        void onOpenChannel(const muduo::net::TcpConnectionPtr &conn, const openChannelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onOpenChannel] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            return cp->openChannel(message);
        }
        void onCloseChannel(const muduo::net::TcpConnectionPtr &conn, const closeChannelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onCloseChannel] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            return cp->closeChannel(message);
        }

        void onDeclareExchange(const muduo::net::TcpConnectionPtr &conn, const declareExchangeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onDeclareExchange] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onDeclareExchange] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->declareExchange(message);
        }

        void onDeleteExchange(const muduo::net::TcpConnectionPtr &conn, const deleteExchangeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onDeleteExchange] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onDeleteExchange] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->deleteExchange(message);
        }

        void onDeclareQueue(const muduo::net::TcpConnectionPtr &conn, const declareQueueRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onDeclareQueue] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onDeclareQueue] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->declareQueue(message);
        }

        void onDeleteQueue(const muduo::net::TcpConnectionPtr &conn, const deleteQueueRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onDeleteQueue] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onDeleteQueue] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->deleteQueue(message);
        }
        void onQueueBind(const muduo::net::TcpConnectionPtr &conn, const BindRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onQueueBind] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onQueueBind] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->queueBind(message);
        }
        void onQueueUnBind(const muduo::net::TcpConnectionPtr &conn, const unBindRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onQueueUnBind] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onQueueUnBind] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->queueUnBind(message);
        }
        void onBasicPublish(const muduo::net::TcpConnectionPtr &conn, const basicPublishRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onBasicPublish] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onBasicPublish] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->basicPublish(message);
        }
        void onBasicAck(const muduo::net::TcpConnectionPtr &conn, const basicAckRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onBasicAck] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onBasicAck] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->basicAck(message);
        }
        void onBasicConsume(const muduo::net::TcpConnectionPtr &conn, const basicConsumeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onBasicConsume] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onBasicConsume] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->basicConsume(message);
        }
        void onBasicCancel(const muduo::net::TcpConnectionPtr &conn, const basicCancelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ConnectionPtr cp = _connection_manager->getConnection(conn);
            if (cp.get() == nullptr)
            {
                DLOG("[onBasicCancel] 连接不存在:%p", conn.get());
                conn->shutdown();
                return;
            }
            Channel::ChannelPtr channel = cp->getChannel(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("[onBasicCancel] 信道不存在:%s", message->cid().c_str());
                conn->shutdown();
                return;
            }
            return channel->basicCancel(message);
        }

    private:
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected() == true)
            {
                _connection_manager->newConnection(_host, _consumer_manager, _codec, conn, _pool);
            }
            else
            {
                _connection_manager->delConnection(conn);
            }
        }

    private:
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        VirtualHost::VirtualHostPtr _host;
        ConsumerManager::ConsumerManagerPtr _consumer_manager;
        ConnectionManager::ConnectionManagerPtr _connection_manager;
        ThreadPool::ThreadPoolPtr _pool;
    };
};